package io.kiki.sba.registry.server.clientInterface.converter.pb;


import com.google.protobuf.UnsafeByteOperations;
import io.kiki.sba.registry.common.model.client.pb.*;
import io.kiki.sba.registry.compress.*;
import io.kiki.sba.registry.core.model.*;
import io.kiki.sba.registry.util.SystemUtils;
import lombok.NonNull;
import org.springframework.util.CollectionUtils;

import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;


public final class ReceivedDataConvertor {
    private static final String KEY_COMPRESS_PUSH_CACHE_CAPACITY = "registry.compress.push.capacity";
    public static final CompressCachedExecutor<CompressedItem> pushCompressExecutor = CompressUtils.newCachedExecutor("compress_push", 60 * 1000, SystemUtils.getSystemInteger(KEY_COMPRESS_PUSH_CACHE_CAPACITY, 1024 * 1024 * 384));

    private ReceivedDataConvertor() {
    }

    public static ReceivedData convert2Java(ReceivedDataPb receivedDataPb) {

        if (receivedDataPb == null) {
            return null;
        }

        ReceivedData receivedData = new ReceivedData();

        receivedData.setData(DataBoxConvertor.convert2JavaMaps(receivedDataPb.getDataMap()));
        receivedData.setDataId(receivedDataPb.getDataId());
        receivedData.setGroup(receivedDataPb.getGroup());
        receivedData.setInstanceId(receivedDataPb.getInstanceId());
        receivedData.setLocalZone(receivedDataPb.getLocalZone());
        receivedData.setScope(receivedDataPb.getScope());
        receivedData.setSegment(receivedDataPb.getSegment());
        receivedData.setSubscriberRegistIds(ListStringConvertor.convert2Java(receivedDataPb.getSubscriberRegistIdsList()));
        receivedData.setVersion(receivedDataPb.getVersion());

        return receivedData;
    }

    public static MultiReceivedData convert2MultiJava(MultiReceivedDataPb multiReceivedDataPb) {
        if (multiReceivedDataPb == null) {
            return null;
        }
        MultiReceivedData multiReceivedData = new MultiReceivedData();
        multiReceivedData.setDataId(multiReceivedDataPb.getDataId());
        multiReceivedData.setInstanceId(multiReceivedDataPb.getInstanceId());
        multiReceivedData.setGroup(multiReceivedDataPb.getGroup());
        multiReceivedData.setScope(multiReceivedDataPb.getScope());
        multiReceivedData.setSubscriberRegistIds(ListStringConvertor.convert2Java(multiReceivedDataPb.getSubscriberRegistIdsList()));
        multiReceivedData.setLocalSegment(multiReceivedDataPb.getLocalSegment());
        multiReceivedData.setLocalZone(multiReceivedDataPb.getLocalZone());

        if (CollectionUtils.isEmpty(multiReceivedDataPb.getMultiDataMap())) {
            multiReceivedData.setMultiData(Collections.EMPTY_MAP);
        } else {
            Map<String, MultiSegmentData> multiData = new HashMap<>(multiReceivedDataPb.getMultiDataMap().size());
            for (Entry<String, MultiSegmentDataPb> entry : multiReceivedDataPb.getMultiDataMap().entrySet()) {
                MultiSegmentDataPb multiSegmentDataPb = entry.getValue();
                MultiSegmentData multiSegmentData = new MultiSegmentData();
                multiSegmentData.setSegment(multiSegmentDataPb.getSegment());
                multiSegmentData.setUnzipData(DataBoxConvertor.convert2JavaMaps(multiSegmentDataPb.getUnzipDataMap()));
                multiSegmentData.setEncoding(multiSegmentDataPb.getEncoding());
                multiSegmentData.setVersion(multiSegmentDataPb.getVersion());
                multiSegmentData.setDataCount(multiSegmentDataPb.getPushDataCountMap());
                multiData.put(entry.getKey(), multiSegmentData);
            }

            multiReceivedData.setMultiData(multiData);
        }
        return multiReceivedData;
    }

    public static MultiReceivedDataPb convert2MultiPb(MultiReceivedData multiReceivedData, CompressorGetter compressorGetter) {
        if (multiReceivedData == null) {
            return null;
        }

        String dataId = multiReceivedData.getDataId();
        String instanceId = multiReceivedData.getInstanceId();
        String group = multiReceivedData.getGroup();
        try {
            MultiReceivedDataPb.Builder builder = MultiReceivedDataPb.newBuilder();
            builder.setDataId(dataId).setGroup(group).setInstanceId(instanceId).setScope(multiReceivedData.getScope()).setLocalSegment(multiReceivedData.getLocalSegment()).setLocalZone(multiReceivedData.getLocalZone()).addAllSubscriberRegistIds(multiReceivedData.getSubscriberRegistIds());

            for (Entry<String, MultiSegmentData> segmentEntry : multiReceivedData.getMultiData().entrySet()) {
                String segment = segmentEntry.getKey();
                MultiSegmentData segmentData = segmentEntry.getValue();

                MultiSegmentDataPb.Builder segmentDataBuilder = MultiSegmentDataPb.newBuilder();
                segmentDataBuilder.setSegment(segmentData.getSegment()).setVersion(segmentData.getVersion()).putAllPushDataCount(segmentData.getDataCount());

                Compressor compressor = compressorGetter.get(segmentData.getUnzipData());
                if (compressor == null) {
                    Map<String, DataBoxesPb> dataBoxesPbMap = DataBoxConvertor.convert2PbMaps(segmentData.getUnzipData());
                    segmentDataBuilder.putAllUnzipData(dataBoxesPbMap);
                } else {
                    CompressedItem compressedItem = dataCenterCompressed(segment, dataId, instanceId, group, segmentData.getVersion(), segmentData.getUnzipData(), compressor);
                    segmentDataBuilder.setZipData(UnsafeByteOperations.unsafeWrap(compressedItem.getCompressedBytes())).setEncoding(compressedItem.getEncoding());
                }
                builder.putMultiData(segment, segmentDataBuilder.build());
            }
            return builder.build();
        } catch (Throwable e) {
            throw new IllegalStateException(e);
        }
    }

    public static CompressedItem dataCenterCompressed(String segment, String dataId, String instanceId, String group, long version, Map<String, List<Data>> zoneData, Compressor compressor) throws Exception {
        CompressedItem compressedItem = pushCompressExecutor.execute(CompressPushKey.of(segment, dataId, instanceId, group, version, zoneData, compressor.getEncoding()), () -> {
            Map<String, DataBoxesPb> dataBoxesPbMap = DataBoxConvertor.convert2PbMaps(zoneData);
            ReceivedDataBodyPb bodyPb = ReceivedDataBodyPb.newBuilder().putAllData(dataBoxesPbMap).build();
            byte[] bodyData = bodyPb.toByteArray();
            byte[] compressed = compressor.compress(bodyData);
            return new CompressedItem(compressed, bodyData.length, compressor.getEncoding());
        });
        return compressedItem;
    }

    public static ReceivedDataPb convert2Pb(ReceivedData receivedDataJava, @NonNull CompressorGetter compressorGetter) {
        if (receivedDataJava == null) {
            return null;
        }


        try {
            Compressor compressor = compressorGetter.get(receivedDataJava.getData());
            String segment = receivedDataJava.getSegment();
            String dataId = receivedDataJava.getDataId();
            String instanceId = receivedDataJava.getInstanceId();
            String group = receivedDataJava.getGroup();

            ReceivedDataPb.Builder builder = ReceivedDataPb.newBuilder();
            builder.setDataId(receivedDataJava.getDataId()).setGroup(receivedDataJava.getGroup()).setInstanceId(receivedDataJava.getInstanceId()).setLocalZone(receivedDataJava.getLocalZone()).setScope(receivedDataJava.getScope()).setSegment(receivedDataJava.getSegment()).setVersion(receivedDataJava.getVersion()).addAllSubscriberRegistIds(receivedDataJava.getSubscriberRegistIds()).putAllPushDataCount(receivedDataJava.getDataCount());

            if (compressor == null) {
                Map<String, DataBoxesPb> dataBoxesPbMap = DataBoxConvertor.convert2PbMaps(receivedDataJava.getData());
                builder.putAllData(dataBoxesPbMap);
            } else {
                CompressedItem compressedItem = dataCenterCompressed(segment, dataId, instanceId, group, receivedDataJava.getVersion(), receivedDataJava.getData(), compressor);
                // compressed data is immutable, use unsafe wrap to avoid copy
                builder.setEncoding(compressedItem.getEncoding()).setBody(UnsafeByteOperations.unsafeWrap(compressedItem.getCompressedBytes())).setOriginBodySize(compressedItem.getOriginSize());
            }

            return builder.build();
        } catch (Throwable e) {
            throw new IllegalStateException(e);
        }
    }

    public static ReceivedConfigDataPb convert2Pb(ReceivedConfigData receivedConfigData) {

        if (receivedConfigData == null) {
            return null;
        }

        ReceivedConfigDataPb.Builder builder = ReceivedConfigDataPb.newBuilder();

        builder.setDataId(receivedConfigData.getDataId()).setGroup(receivedConfigData.getGroup()).setInstanceId(receivedConfigData.getInstanceId()).setVersion(receivedConfigData.getVersion()).addAllConfiguratorRegistIds(receivedConfigData.getConfiguratorRegistIds()).setDataBox(DataBoxConvertor.convert2Pb(receivedConfigData.getData()));

        return builder.build();
    }

    public interface CompressorGetter {
        Compressor get(Map<String, List<Data>> data);
    }
}
